package com.xs.mvc.common;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

import com.alibaba.fastjson.JSON;
import com.xs.mvc.web.BaseController;

/**
 * @Vitle: <P>KafkaUtil.java</P>
 * @Package <P>com.xs.mvc.common</P>
 * @Description: <P>KafkaSendMsgUtils</P>
 * @author renweiping
 * @date 2019年3月12日 上午9:45:05
 * @version 1.0
 */
@Component
public class KafkaSendMsgUtils<K,V> {
	
	public final Logger LOG = Logger.getLogger(KafkaSendMsgUtils.class);
	
	@SuppressWarnings("rawtypes")
	@Autowired
	private  KafkaTemplate kafkaTemplate;


	@SuppressWarnings("unchecked")
	public  <K,V>void sendMessage(String topic, Integer partition, Long timestamp,  K key, V data) {
       
        ListenableFuture<SendResult<K, V>> listenableFuture = null;
        if (kafkaTemplate.getDefaultTopic().equals(topic)) {
            listenableFuture =  kafkaTemplate.sendDefault(partition,timestamp,key,data);
        }else {
        	ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic,partition,timestamp, key, data);
            listenableFuture = kafkaTemplate.send(producerRecord);
        }
        //发送成功回调
        SuccessCallback<SendResult<K, V>> successCallback = new SuccessCallback<SendResult<K, V>>() {
            @Override
            public void onSuccess(SendResult<K, V> result) {
                LOG.info("***************发送数据：{\"topic\"="+topic+",\"partition\"="+partition+",\"timestamp\"="+timestamp+","
                		+ "\"key\"="+key+",\"data\"="+data+"}发送成功");
            }
        };
        //发送失败回调
        FailureCallback failureCallback = new FailureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                //失败业务逻辑
            	LOG.error("KafkaSendMsgUtils.sendMessage()发生异常："+ex);
                throw new RuntimeException(ex);
            }
        };
        listenableFuture.addCallback(successCallback, failureCallback);
    }

}
